package com.nl.utils;

import com.nl.bean.output.WindowAggResult;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 *Description:自定义窗口函数，将窗口聚合结果包装成WindowAggResult输出
 *  ProcessWindowFunction<IN, OUT, KEY, W extends Window>
 *  IN:输入也就是预聚合函数的输入,这边和CountAgg配合使用就是long
 *  OUT:输出类型输出WindowAggResult
 *  KEY:分组key类型,如果keyBy不是用KeySelector这边是Tuple。这边用泛型
 *  W:TimeWindow
 *Version:1.0.0
 *@author shihb
 *@date 2020/4/13 9:01
 */
public class WindowAggResultOut<KEY> extends
    ProcessWindowFunction<Long, WindowAggResult, KEY, TimeWindow> {

  @Override
  public void process(KEY key, Context context, Iterable<Long> elements,
      Collector<WindowAggResult> out) throws Exception {
    out.collect(new WindowAggResult<KEY>(key, context.window().getEnd(), elements.iterator().next()));
  }
}
